
----------------------------------------------------------------
--- Copyright：  迷你玩-海马
--- Description：（只提供给 data_cache服务 ）
--- dc读写mysql的封装，【其他业务应该不会使用到这个文件】
---              请求kv云数据库模块，带有share_dict共享内存缓存
---
---              接入 dc-mysql请使用 mysqlDcClient.lua
---              纯 mysql 查询请使用 mysqlMgr.lua
----------------------------------------------------------------


local assert     = assert
local require    = require    
local bit        = bit
local coroutine  = coroutine
local debug      = debug
local io         = io
local pairs      = pairs
local ipairs     = ipairs
local math       = math
local os         = os
local print      = print
local pcall      = pcall
local xpcall     = xpcall
local rawget     = rawget
local rawset     = rawset
local select     = select
local string     = string
local table      = table
local tonumber   = tonumber
local tostring	 = tostring
local error      = error
local type       = type
local unpack     = unpack
local setmetatable = setmetatable
local getmetatable = getmetatable
local ngx = ngx

local table_insert = table.insert
local table_remove = table.remove
local ngx_sleep = ngx.sleep
local ngx_now   = ngx.now
local table_concat = table.concat

require "resty.core.shdict"

local ngx_shared  = ngx.shared;
local string_sub  = string.sub
local string_gsub = string.gsub
local string_format = string.format

local log       = require("modules.commhm.log");
local ns_time   = require("modules.commhm.time");
local uu        = require("modules.commhm.utils");
local ns_env    = require("modules.env.env");

local ns_mysql     = require("resty.mysql");
local ns_semaphore = require("ngx.semaphore")

local __TJ_INTERVAL__       = 180    ---统计间隔

local __SELECT_COUNT__      =  10       ---sema_list 分表数，一般为10 (mysql 独有 )


local __BUSY_BLOCK_LIMIT__     = 64       ---限流次数

local __SEMA_RESOURCES__       =  5         -- 最大并发赛道
local __SEMA_PROTECT_BREAK__   = (0-5*64)   -- 熔断保护值 (同时block mread read save)



local NGX_NULL  = "not_found"       --ngx.null
local __VER__   = '__VeR_'          --数据版本后缀，大小写防止冲突


local __MYSQL_TIME_OUT_LOG__ = 0.1    --mysql超时打印日志时间
if  ns_env.debug == 1 then
	__MYSQL_TIME_OUT_LOG__ = 0        --测试环境一定打印
end


-- MYSQL MGR类
local _M = { _VERSION = '0.12.1' }
local mt = { __index = _M }


--- params_: 附加参数 { keepalive_pool_size=5 }
function _M.new(self, pre_, config_server_, config_select_, params_ )
	log.debug( log.fast() and "call mysqlMgr:new" );
	local ins_ = {
		config_cache  = {};
		pre           = pre_;
		config_server = config_server_;
		config_select = config_select_;

		select_mod    = (config_select_.select_mod or 10);  ---10分表  100分表

		last_time_report_tj        = uu.now() + __TJ_INTERVAL__;

		mysql_delay_clear = {};

		get_set_tj   = {};
		-- get_call
		-- get_cache
		-- get_cache_stale
		-- get_mysql
		-- set_mysql
		-- delay_mysql

		mread_tj = {
			call  = 0,
			get   = 0,
			gets  = 0,
			set   = 0,
		};

		_lock_id   = (ngx.worker.id() + 1) * 1000;    --lock_id

		read_only_ssdb_block = 0,    --忙的时候流控 无法mread
		socket_link_count = 0,       --当前连接数(并发任务数)
		sema_list = {},              --信号锁

		keepalive_pool_size = 16;
	};

	if  params_ then
		if  params_.keepalive_pool_size then
			ins_.keepalive_pool_size = params_.keepalive_pool_size
		end
	end

    return setmetatable(ins_, mt);
end


_M.final = function(self)
	self:do_report_tj();
end


---计算lock_expire_time的合法性
local check_lock_expire_time = function( p_et_ )
	local et_ = 3000
	if  p_et_ then
		et_ = tonumber( p_et_ ) or 0
		if  et_ < 1000 or et_ > 60000 then   --- 最小1秒，最大60秒
			et_ = 3000
		end
	end
	return et_
end


----执行mysql----
---sql        : sql语句
---select_    : 分库分表的数字
_M.query = function(self, sql, select_ )
	log.debug( log.fast() and "sql=" .. sql .. ", select=" .. select_ )

	if  not ( 0 == self:wait_sema( select_ ) ) then
		return false, nil, "FAIL_sema"
	end

    local instance, err = self:try_connect(select_ );
	if  not instance then
		self:post_sema( select_ )    --连接失败 释放sema
		return false, nil, err;
	end

	local start_ = ngx.now()
    local res, err, errcode, sqlstate = instance:query(sql);

	self:try_close( instance, select_, err )
	self:post_sema( select_ )  --释放sema

	local cost_ = ngx.now() - start_
	if  cost_ >= __MYSQL_TIME_OUT_LOG__ then   --100ms
		log.day_list_simple( "dc_mysql_cost", ngx.worker.id(), "query",
				math.floor(cost_*1000), string.sub(sql, 1, 128)..'...' )
	end

	if  err or errcode then
		log.day_list( "dc_mysql", "error_errcode", select_, err, errcode, sqlstate, sql )
	end

    if  not res then
		log.day_list( "error", "dc_mysql", select_, err, errcode, sqlstate, sql )
		return false, nil, err;
	else
		log.debug( log.fast() and "call mysql query ok=true" );
		uu.var_dump( res );
		return true, res, err;
    end
end


-------------------------------------------------------
-- msg: 返回给调用方的错误描述
-- err: 详细错误
_M.call_mysql_fail = function(self, msg, key, hash, err)
	log.day_list_simple( "dc_mysql", "error", msg, key, hash, err )
	self:close(true)

	if  ngx.ctx.err_call_back then
		ngx.ctx.err_call_back( err )
	end
end



_M.private_close = function( self, instance, select_, shutdown )
	local ok, err;
	if  shutdown == true  then
		ok, err = instance:close();
		log.day_list( "error", "dc_mysql", select_, "close" )
	else
		ok, err = instance:set_keepalive( 120000, self.keepalive_pool_size );   --2min
		--log.day_list( "dc_mysql" , "set_keepalive", select_, ngx.worker.id(), 0, self.keepalive_pool_size )
	end

	self.socket_link_count = self.socket_link_count - 1
	if  self.socket_link_count < 0 or self.socket_link_count > 1 then
		log.day_list_short( "dc_mysql", ngx.worker.id(), "socket_link_count", self.socket_link_count, self.pre )
	end

	if  not ok then
    	log.debug( log.fast() and "failed to set connection mysql keepalive: "
				.. (err or "nil") .. ", select=" .. select_ );
    else
    	log.debug( log.fast() and "dc_mysql connection keep alive ok. select=" .. select_ );
    end
end



---------------------------------------------------------------
-- 按分库分表配置获得tv的表名
_M.private_get_kv_table_name = function(self, select_)
	local config_ = self:get_config_by_key( select_ )
	if  config_ and config_.table_name then
		return config_.table_name
	else
		return 't_miniw_kv'
	end
end


-- 按照 config 分表获得连接IP和端口
_M.get_config_by_key = function(self, p_select_)
	log.debug( log.fast() and "get_config_by_key, select=" .. p_select_ );

	local select_ = p_select_ % self.select_mod

	if  self.config_cache[ select_ ]  then
		return  self.config_cache[ select_ ];
	else
		--log.debug( "new mysql config" );
		--uu.var_dump( self.config_server )
		if  self.config_server and self.config_select then
			if  self.config_select[select_] then
				local ret_ = self.config_server[ self.config_select[select_] ];
				if  ret_ then
					log.debug( log.fast() and "set config_cache for " .. select_ );
					self.config_cache[ select_ ] = ret_;
					return ret_;
				end
			end
		end
	end

	self:call_mysql_fail( "error_config", nil, nil, 'config mysql select, error = ' .. (select_ or 'nil') )
end


--获得连接
_M.try_connect = function( self, select_ )
	local config_ = self:get_config_by_key( select_ );

	--新建连接
	log.debug( log.fast() and "create new mysql connection. select=" .. select_ );
	--uu.var_dump( config_ );
	local instance, err = self:private_try_connect( config_, select_ );
	if  not instance then
		return nil, err
	end

	return instance;
end



-- 尝试关闭一个连接
_M.try_close = function( self, instance, select_, err )
	if  err then
		log.error("try_close: " .. uu.to_str(err) .. ", select=" .. select_ )
		self:private_close( instance, select_, true )   --shutdown socket
	else
		self:private_close( instance, select_ )         --keep alive socket
	end
end



-----------------------------------------------------------------------------utils---------------------
-- config = { ip, port, user, pass, select }
_M.private_try_connect = function( self, config_, select_ )

	local instance = ns_mysql:new();
	instance:set_timeout(3000);      --3s

	--log.debug( log.fast() and "set pool name=" .. config_.pool );

	local ok, err, errcode, sqlstate = instance:connect( config_ );
	if  not ok then
		log.day_list( "error", "dc_mysql", select_, err, errcode, sqlstate, "fail_connect"  );
		return nil, "FAIL_connect"
	end

	--log.day_list( "socket_mysql", ngx.worker.id(), '-', uu.to_str(config_),
			--tostring(instance), tostring(instance.sock), instance:get_reused_times() )

	local count, err = instance:get_reused_times();
	if  err then
		log.day_list( "error", "dc_mysql", select_, err, "fail_get_reused_times"  );
		return nil, "FAIL_get_reused_times"
	else
		log.debug( log.fast() and "dc_mysql get_reused_times=" .. count );
	end

	self.socket_link_count = self.socket_link_count + 1
	return instance;
end



-------------------------------
-- 等待资源锁
_M.wait_sema = function( self, p_select_ )
	local select_ = p_select_ % __SELECT_COUNT__    -- 最大支持10队列

	if  not self.sema_list[ select_ ] then
		--sema=信号锁  t=当前秒  link=当前秒的赛道并行数 max=当前秒出现的赛道最大值
		self.sema_list[ select_ ] = { sema = ns_semaphore.new( __SEMA_RESOURCES__ ), t=0, link=0, max=0 }
	end

	local sema_ = self.sema_list[ select_ ].sema

	if  sema_:count() < __SEMA_PROTECT_BREAK__ then
		-- 资源数耗尽，队列过长
		log.day_list_short( "dc_mysql_cost", ngx.worker.id(), "wait_list_limit",
				self.pre, select_, sema_:count(), self.socket_link_count )

		self.read_only_ssdb_block = __BUSY_BLOCK_LIMIT__   ---触发限流
		return 1
	end


	local start_  = ngx.now()
	local ok, err = sema_:wait(2)   --最大等待2秒
	local cost_   = ngx.now() - start_
	if  ok then

		-- link +1
		self.sema_list[ select_ ].link = self.sema_list[ select_ ].link + 1
		if  self.sema_list[ select_ ].max < self.sema_list[ select_ ].link then
			self.sema_list[ select_ ].max = self.sema_list[ select_ ].link
		end

		if  cost_ >= 0.001 then   --10ms
			log.day_list_short( "dc_mysql_cost", ngx.worker.id(), "wait_ok", self.pre, select_,
					math.floor(cost_*1000),
					'sema=' .. sema_:count(),
					'max='..self.sema_list[ select_ ].max,
					'link='..self.sema_list[ select_ ].link,
					'slink='..self.socket_link_count )
		end

		return 0  ---正常唤醒 拿到锁
	else
		-- 一般都是 timeout
		log.day_list_short( "dc_mysql_cost", ngx.worker.id(), "wait_timeout", self.pre, select_,
				err,  math.floor(cost_*1000),
				'sema=' .. sema_:count(),
				'max='..self.sema_list[ select_ ].max,
				'link='..self.sema_list[ select_ ].link,
				'slink='..self.socket_link_count )

		log.error("ERROR: read_semaphore " .. err)
	end
	return 1
end


-- 释放锁
_M.post_sema = function( self, p_select_ )
	local select_ = p_select_ % __SELECT_COUNT__    -- 最大支持10队列
	if  self.sema_list and self.sema_list[ select_ ] then
		local sema_info_ = self.sema_list[ select_ ]

		-- thread减1
		self.sema_list[ select_ ].link = self.sema_list[ select_ ].link - 1

		-- 转秒计数清0
		if  sema_info_.t < ngx.time() then
			sema_info_.t = ngx.time()
			if  sema_info_.link ~= 0 then
				log.day_list_short( "dc_mysql_cost", ngx.worker.id(), "sema_max", self.pre, select_,
						'sema=' .. sema_info_.sema:count(),
						'max='..self.sema_list[ select_ ].max,
						'link='..self.sema_list[ select_ ].link,
						'slink='..self.socket_link_count )
			end
			sema_info_.max = 0
		end

		if  sema_info_.sema:count() < __SEMA_RESOURCES__ then
			sema_info_.sema:post(1)   --可用资源数+1
		else
			-- 超限(不用拉起) 应该不会发生
			log.day_list_short( "dc_mysql_cost", ngx.worker.id(), "error_sema_bigger_res", self.pre, select_,
					sema_info_.sema:count() )
		end
	end
end


-------------------------------------------------------------------------------

_M.set_get_set_tj = function(self, hash_, name_)
	if     ngx.ctx.m_params.act then
		hash_ = hash_ .. "_" .. ngx.ctx.m_params.act
	elseif ngx.ctx.m_params.cmd then
		hash_ = hash_ .. "_" .. ngx.ctx.m_params.cmd
	end

	if  not self.get_set_tj[hash_] then
		self.get_set_tj[hash_] = {
			get_call  = 0,
			get_cache = 0,
			get_cache_stale  = 0,
			get_mysql = 0,
			set_mysql = 0,
			delay_mysql = 0,
		};
	end

	if  self.get_set_tj[hash_][ name_ ] then
		self.get_set_tj[hash_][ name_ ] = self.get_set_tj[hash_][ name_ ] + 1
	else
		log.error( "call set_get_set_tj, unknown name_=" .. name_ )
	end

end


_M.do_report_tj = function(self)
	local now_ = uu.now()
	if  now_ > self.last_time_report_tj then
		self.last_time_report_tj = now_ + __TJ_INTERVAL__;

		--开始打印统计
		for hash_, v1 in pairs(self.get_set_tj) do
			if  v1.get_call + v1.set_mysql > 0 then
				log.day_list_short( "dc_mysql", "cache_hit_mysql", "get_set", self.pre,
						"get", v1.get_call, v1.get_cache,  v1.get_cache_stale, v1.get_mysql,
						"set", v1.set_mysql, v1.delay_mysql, hash_ )
			end
		end
		self.get_set_tj = {}

		if  self.mread_tj.call + self.mread_tj.get + self.mread_tj.set > 0 then
			log.day_list_short( "dc_mysql", "cache_hit_mysql", "mread", self.pre, self.mread_tj.call,
					self.mread_tj.get, self.mread_tj.gets, self.mread_tj.set )
			self.mread_tj = { call=0, get=0, gets=0, set=0 }
		end
	end
end



---进行mysql key_hash锁检查  0=失败   大于0=成功
_M.check_mysql_resty_lock = function(self, select_, key_hash_, et_ )
	local dict_ = ngx_shared[self.pre .. select_]
	if  dict_ then
		--N秒中内循环试图锁定一个 key_hash_lk ( exists则等待 )
		local timeout_     = et_              ---毫秒
		local dict_timeout = et_ * 0.001 + 1  ---秒 ( dict超时比wait一定要长 )

		local step_    = 1      ---步长毫秒
		local cc_      = 0

		---计算锁id
		self._lock_id = self._lock_id + 1
		if  self._lock_id >= 100000 then self._lock_id = 1000 end

		local return_when_lock_ = ngx.ctx.m_params.return_when_lock   --是否有锁立即返回

		while timeout_ > 0 do
			cc_ = cc_ + 1

			local ok, err = dict_:add( key_hash_ .. "_lk", self._lock_id, dict_timeout )      ---锁住主键kh  12345_6_lk
			if  ok then
				if  cc_ > 10 then
					--循环等待其他锁的次数太多
					log.day_list_short( "dc_mysql", "lock_rw", "lock_ok0", ngx.worker.id(), uu.get_act_cmd(),
							key_hash_, cc_, et_, et_-timeout_, step_, self._lock_id );
				end
				return self._lock_id   --成功
			else

				if  err == 'exists' then
					if  return_when_lock_ == '1' then
						return 0, "locked"  --是否遇到锁立即返回
					end

					timeout_ = timeout_ - step_
					if  timeout_ <= 0 then
						log.day_list_short("dc_mysql", "lock_rw", "error_time_out0",
								ngx.worker.id(), uu.get_act_cmd(),
								key_hash_, cc_, et_, timeout_, step_, debug.traceback() )
						return 0, "error_time_out_lock"    --超时
					end

					if  cc_ <= 10 then
						step_ = 50 + ( cc_ * 2)   ---等待时间逐步放大   500+2+4+6+8+10+12+14+16+18+20=500 + 110
					else
						step_ = (20-cc_) * 10      ---等待时间逐步缩小  100 90 80 70 60 50 40 30 20
						if  step_ < 20 then step_ = 20 end
					end

					--log.day_list_short( "dc_mysql", "lock_rw", "step0", ngx.worker.id(), uu.get_act_cmd(),
						--key_hash_, timeout_, step_, debug.traceback() );
					ngx_sleep( step_ * 0.001 )

				else
					log.day_list_short( "dc_mysql", "lock_rw", "error_step0",
							ngx.worker.id(), uu.get_act_cmd(), self.pre, select_,
							key_hash_, et_, cc_, timeout_, step_, err, debug.traceback() );
					return  0, err  	---其他错误
				end
			end
		end

	else
		if  self.select_mod ~= 10 then
			---提示可能是百分表
			log.error( "error_select: plz check select_mod=" .. self.select_mod )
		end

		log.day_list_short("dc_mysql", "lock_rw", "error_no_select", ngx.worker.id(), uu.get_act_cmd(),
				self.pre, select_, key_hash_, debug.traceback() );
		return 0, "error_no_ngx_shared"   ---找不到 ngx_shared
	end

	return 0, "unknown_err"   ---未知错误 (应该不能运行到这里，健壮性保证)
end


---远程调用解锁，只有解锁后，其他协程才能继续获得读写锁
---read_only_ 不解锁，等待final统一解锁
---返回值1 0=成功
---返回值2 msg=错误描述
_M.unlock_resty_lock = function( self, select_, key_hash_, read_only_ )
	if  read_only_ then
		return 0, "ok"  ---不解锁
	end

	local dict_ = ngx_shared[ self.pre .. select_ ]
	if  dict_ then

		if  ngx.ctx.m_params.lock_id then
			---  检查lock_id  (普通read save锁)
			local lock_id_ = dict_:get( key_hash_ .. "_lk" )
			if  lock_id_ then
				if  lock_id_ == tonumber(ngx.ctx.m_params.lock_id) then
					--log.day_list_simple( "dc_mysql", "unlock_check_lock_id_ok",   key_hash_, lock_id_ )
				else
					log.day_list_simple( "dc_mysql", "unlock_check_lock_id_fail", key_hash_, lock_id_, ngx.ctx.m_params.lock_id)
					return 1, "lock_id_not_match"
				end
			end

		elseif  ngx.ctx.m_params.lock_kh_id then
			---  检查 lock_kh_id (通过lock_kh获得的远程分布式锁)
			local lock_id_ = dict_:get( key_hash_ .. "_lk" )
			if  lock_id_ then
				if  lock_id_ == tonumber(ngx.ctx.m_params.lock_kh_id) then
					--log.day_list_simple( "dc_mysql", "unlock_check_lock_kh_id_ok",   key_hash_, lock_id_ )
				else
					log.day_list_simple( "dc_mysql", "unlock_check_lock_kh_id_fail", key_hash_, lock_id_, ngx.ctx.m_params.lock_kh_id)
					return 1, "lock_kh_id_not_match"
				end
			end
		end

		local ok, err_ = dict_:delete( key_hash_ .. "_lk")
		if  ok then
			--删除成功
			--log.day_list_short( "dc_mysql", "lock_rw", "unlock_ok", ngx.worker.id(), uu.get_act_cmd(), key_hash_ );
			return 0, "ok"
		else
			log.day_list_short( "dc_mysql", "lock_rw", "error_delete_unlock1", ngx.worker.id(),
					uu.get_act_cmd(), key_hash_, err_, debug.traceback() );
			return 2, "not_lock"
		end
	else
		log.error( "dict_not_exist=" .. self.pre .. select_ )
		log.day_list_short( "dc_mysql", "lock_rw", "error_dict_not_exist",  self.pre .. select_ );
		return 3, "no_dict"
	end

end



-------------------------------------------------------------------------------
--- API 读请求 ---
_M.API_mysql_read = function( self, p_select_, key, hash, err_call_back, read_only_, et_ )
	local key_str_   = uu.check_sql_field_quote( key )

	local select_ = p_select_ % self.select_mod
	ngx.ctx.err_call_back = err_call_back;

	self:set_get_set_tj( hash, "get_call" )
	local key_hash_ = key .. "_h" .. hash

	--乐观锁r
	local lock_kh_id_ = ngx.ctx.m_params.lock_kh_id
	if  read_only_ == '1' then
		--只读接口
	elseif lock_kh_id_ then
		--使用lock_kh的锁
		local ret_, msg_ = self:check_lock_kh_id( select_, key_hash_, lock_kh_id_ )
		if  ret_ ~= 0 then
			return ret_, msg_   --校lock验锁失败
		end
	else
		local lock_id_, err_msg_ = self:check_mysql_resty_lock(select_, key_hash_, check_lock_expire_time(et_) )
		if  lock_id_ == 0 then
			---获得锁失败 一般是超时 read请求并没有发出去，也没有开始connect，不断开其他链接
			self:call_mysql_fail( "check_read_lock", key, hash, err_msg_ )
			return 500, "check_read_lock"
		end
		ngx.ctx.__lock_id__ = lock_id_   ---记录lock_id
	end

	local ret2, stale_ = self:get_shared_cache_mysql( select_, key_hash_ )
	if  ret2 then
		--log.debug( log.fast() and "ret2 cache hit." )
		if  stale_ then
			self:set_get_set_tj( hash, "get_cache_stale" )
		else
			self:set_get_set_tj( hash, "get_cache" )
		end
		return 0, ret2   --cache命中
	end

	local ret = NGX_NULL;

	----id k1 hash v1 ver
	local mysql_ = table_concat( { "select ver, UNCOMPRESS(v1) as v1 from ", self:private_get_kv_table_name(select_),
			" where k1=", key_str_, " and hash=", hash, " limit 1" } )

	local ret_, ret_table_, err = self:query( mysql_, select_ )
	if  not ret_ then
		self:call_mysql_fail( "select_fail", key, hash, err )
		return 500, "select_fail"
	end

	-- 处理返回值 返回是字符串或者NGX_NULL
	if  type(ret_table_) == 'table' then
		if  ret_table_[1] and ret_table_[1].v1 then
			ret = ret_table_[1].v1
			if  ngx.null == ret then
				ret = ""
			end
			self:set_get_set_tj( hash, "get_mysql" )
			if  ngx.ctx.no_shared_cache == 'y' then
				--不保存 scan使用
				log.day_list_short( "dc_mysql", "no_shared_cache", select_, key, hash );
			else
				if #ret > 0 then
					self:set_shared_cache_mysql( select_, key_hash_, ret, nil, ret_table_[1].ver )
				else
					log.day_list_simple( "dc_mysql", "error_len0", select_, key, hash, ret );
				end
			end
		else
			---空数据
		end
		return 0, ret
	else
		-- 返回的一定是一个table 否则是出错
		self:call_mysql_fail( "ret_not_table", key, hash, err )
		return 500, 'ret_not_table'
	end

end



-------------------------------------------------------
--- API 写请求
_M.API_mysql_save = function( self, p_select_, key, hash, value, p_ver_, err_call_back, mysql_delay_, read_only_ )
	local select_   = p_select_ % self.select_mod
	local key_hash_ = key .. "_h" .. hash

	local lock_kh_id_ = ngx.ctx.m_params.lock_kh_id
	if  read_only_ == '1' then
		-- 无锁
	elseif lock_kh_id_ then
		--使用lock_kh的锁
		local ret_, msg_ = self:check_lock_kh_id( select_, key_hash_, lock_kh_id_ )
		if  ret_ ~= 0 then
			return msg_   --校lock验锁失败
		end
	end

	local key_str_   = uu.check_sql_field_quote( key )
	local value_str_ = uu.check_sql_field_quote( value )

	ngx.ctx.err_call_back = err_call_back;

	---- 版本对比，有ver且大于0的时候，需要对比数据库的ver
	local db_ver_  = 0   --数据库中的数据版本
	local ver_     = 0   --输入参数版本
	if  p_ver_ then	ver_ = tonumber( p_ver_ ) or 0 end

	local table_name_ = self:private_get_kv_table_name(select_)

	local mysql_
	if  ver_ > 0 then
		-- ver大于0的时候，需要做版本比对，传入的ver要大于数据库中的ver才能写入
		local string_ver_ = self:get_ver_shared_cache_mysql( select_, key_hash_ )
		if  string_ver_ then
			-- cache中获取到了ver
			db_ver_ = tonumber(string_ver_) or 0
		else
			-- 从数据库中获得ver
			mysql_ = table_concat( { "select ver from ", table_name_, " where k1=", key_str_, " and hash=", hash } )
			local ret_, ret_table_, err = self:query( mysql_, select_ )
			if  not ret_ then
				self:call_mysql_fail( "select_ver", key, hash, err )
				return
			end
			uu.var_dump(ret_table_)
			if  ret_table_ and ret_table_[1] and ret_table_[1].ver then
				db_ver_ = ret_table_[1].ver
			end
		end

		if  not ( db_ver_ + 1 == ver_ ) then
			if  db_ver_ == 0 then
				-- 删除后再写，参数ver可以大于1，比如100，200
				log.day_list( "dc_mysql", "ver_save_after_del", key_str_, hash, db_ver_, ver_ )
			else
				log.day_list( "dc_mysql", "ver_check_fail", key_str_, hash, db_ver_, ver_ )
				return "ver_check_fail"
			end
		end
	end


	-- 优先写入share cache
	local ret_ = self:set_shared_cache_mysql( select_, key_hash_, value, nil, ver_ )
	if  ret_ and tonumber(mysql_delay_) == 1 then
		--缓写落地数据
		self:unlock_resty_lock( select_, key_hash_, read_only_ )
		self:set_get_set_tj( hash, "delay_mysql" )
		return "OK"
	end

	-- check if existed: on duplicate key update
	mysql_ = table_concat( { "insert into ", table_name_,
			"(id,k1,hash,v1,ver,t1) values( null,", key_str_, ",", hash,
			",COMPRESS( ", value_str_, " ),", ver_, ",", uu.now(),
			") on duplicate key update v1=COMPRESS( ", value_str_, " ),ver=",ver_,",t1=", uu.now() } )

	local ret_, ret_table_, err = self:query( mysql_, select_ )
	if  not ret_ then
		-- 请求失败，删除掉cache
		self:clean_shared_cache_mysql( select_, key, hash )
		self:call_mysql_fail( "insert_fail", key, hash, err )
	else
		uu.var_dump(ret_table_)
		if  lock_kh_id_ then
			--有远程分布式锁
		else
			self:unlock_resty_lock( select_, key_hash_, read_only_ )
		end
	end

	self:set_get_set_tj( hash, "set_mysql" )

	return "OK"
end



--- API multi read
_M.API_mysql_mread = function( self, p_select_, all_kv_list, err_call_back )
	local select_ = p_select_ % self.select_mod

	ngx.ctx.err_call_back = err_call_back;

	local fret = {};          ---最后结果
	local fret_stdmap = {};   ---帮助排序结构体

	uu.var_dump( all_kv_list )

	--先查cache
	local kv_list = {}           --需要mysql的list
	for i=1, #all_kv_list do
		self.mread_tj.call = self.mread_tj.call + 1

		local key_hash_ = all_kv_list[i].key .. "_h" .. all_kv_list[i].hash

		fret_stdmap[ key_hash_ ] = all_kv_list[i]     ---fret_stdmap和all_kv_list映射起来 [ table --指针--> table ]

		local ret2, stale_ = self:get_shared_cache_mysql( select_, key_hash_ )
		if  ret2 then
			--命中
			if  stale_ then
				self.mread_tj.gets = self.mread_tj.gets + 1
			else
				self.mread_tj.get  = self.mread_tj.get + 1
			end
			fret_stdmap[ key_hash_ ].v = ret2   ---记录结果   ---设置fret_stdmap等于设置all_kv_list
			if  log.fast() then
				log.debug( "find cache:" .. select_ .. "," .. key_hash_ )
			end
		else
			--需要拉取
			kv_list[ #kv_list + 1 ] = all_kv_list[i]

			-- 只要触发流控，阻止请求ssdb，可以请求shared_cache
			if  self.read_only_ssdb_block > 0 then
				self.read_only_ssdb_block = self.read_only_ssdb_block - 1
				log.day_list_short( "dc_mysql", "mread|read_only_ssdb_block|" .. self.read_only_ssdb_block );
				self:call_mysql_fail( "mread_read_only_ssdb_block", 0, 0, "block" )
				return
			end

		end
	end

	-- 再查mysql
	if  #kv_list > 0 then
		local mysql_ = { "select k1, hash, UNCOMPRESS(v1) as v1 from ",
						  self:private_get_kv_table_name(select_), " where " }
		for i=1, #kv_list do
			if  i > 1 then
				mysql_[ #mysql_ + 1 ] = " or "
			end
			local key_str_ = uu.check_sql_field_quote( kv_list[i].key )
			mysql_[ #mysql_ + 1 ] = table_concat( { "(k1=", key_str_, " and hash=", kv_list[i].hash, ")" } )
		end
		local mysql_str_ = table_concat( mysql_ )

		local ret_, ret_table_, err = self:query( mysql_str_, select_ )
		if  not ret_ then
			self:call_mysql_fail( "mread_select_fail", 0, 0, err )
			return
		end

		if  type(ret_table_) == 'table' then
			for i=1, #ret_table_ do
				local ret
				if  type(ret_table_[i]) == 'table' and ret_table_[i].v1 then
					local key_hash_ = ret_table_[i].k1 .. "_h" .. ret_table_[i].hash
					ret = ret_table_[i].v1
					--ver = ret_table_[i].ver   --mread不会使用ver
					--t1  = ret_table_[i].t1    --mread不会使用t1
					if  ret and type(ret) == 'string' then
						self:set_shared_cache_mysql( select_, key_hash_, ret, 10*60 )
						self.mread_tj.set = self.mread_tj.set + 1
					else
						ret = NGX_NULL
						log.day_list_short( "dc_mysql", "mread|ret_1_not_string",  key_hash_ );
					end

					if  ret and fret_stdmap[ key_hash_ ] then
						fret_stdmap[ key_hash_ ].v   = ret   ---设置fret_stdmap等于设置all_kv_list
						--fret_stdmap[ key_hash_ ].ver = ver   --mread不会使用ver
						--fret_stdmap[ key_hash_ ].t1  = t1    --mread不会使用t1
					end
				else
					-- 返回的一定是一个table 否则是出错
					log.error( "ret_not_table" .. uu.to_str(ret_table_[i]) )
					log.day_list_short( "dc_mysql", "mread|ret_not_table", uu.to_str(ret_table_[i]) )
				end
			end
		end
	end

	--- 最后合并结果
	for i=1, #all_kv_list do
		if  all_kv_list[i] then
			if  not all_kv_list[i].v then
				all_kv_list[i].v = NGX_NULL
			end
			fret[ i ] = all_kv_list[i]	-- 这里和ssdb不同,需要整个结构体，dc_client重新组装 (dc_gate mm_read需要用到)
		end
	end

	return fret;
end


--- API  删除一个key下的一个hash
_M.API_mysql_del = function( self, p_select_, key, hash, err_call_back )
	log.debug( log.fast() and "call API_mysql_del, select_=" .. p_select_
			.. ", key=" .. key .. ", hash=" .. hash );

	local key_str_   = uu.check_sql_field_quote( key )

	local select_ = p_select_ % self.select_mod
	ngx.ctx.err_call_back = err_call_back;

	--self, select_, key_, hash_ )
	self:clean_shared_cache_mysql( select_, key, hash )

	local mysql_ = "delete from " .. self:private_get_kv_table_name(select_)
			.. " where k1=" .. key_str_ .. " and hash=" .. hash .. " limit 1"
	local ret_, ret_table_, err = self:query( mysql_, select_ )
	if  not ret_ then
		self:call_mysql_fail( "delete_fail", key, hash, err )
		return;
	end

	self:unlock_resty_lock( select_, key .. "_h" .. hash )  --key_hash_

	uu.var_dump( ret_table_ );

	if  ret_table_ and ret_table_[1] then
		return ret_table_[1];
	else
		return 0;
	end
end


--- API 一个key下有多少个值，最多1024个
_M.API_mysql_hlen = function( self, p_select_, key, err_call_back)
	log.debug( log.fast() and "call API_mysql_hlen, select_=" .. p_select_ .. ", key=" .. key );

	local key_str_   = uu.check_sql_field_quote( key )

	local select_ = p_select_ % self.select_mod
	ngx.ctx.err_call_back = err_call_back;

	local mysql_ = "select hash from " .. self:private_get_kv_table_name(select_)
			.. " where k1=" .. key_str_ .. " limit 1024"
	local ret_, ret_table_, err = self:query( mysql_, select_ )
	if  not ret_ then
		self:call_mysql_fail( "hlen_fail", key, 0, err )
		return;
	end

	uu.var_dump( ret_table_ );
	if  ret_table_ then
		return #ret_table_
	end
end


--- API 获得一个key下的所有值
_M.API_mysql_hget_all = function(self, p_select_, key, err_call_back)
	log.debug( log.fast() and "call API_mysql_hget_all, key=" .. key );

	local key_str_   = uu.check_sql_field_quote( key )

	local select_ = p_select_ % self.select_mod
	ngx.ctx.err_call_back = err_call_back;

	local mysql_ = "select  k1, hash, UNCOMPRESS(v1) as v1, ver from " .. self:private_get_kv_table_name(select_)
			.. " where k1=" .. key_str_
	local ret_, ret_table_, err = self:query(mysql_, select_);
	if  not ret_ then
		self:call_mysql_fail( "hget_all_fail", key, 0, err )
		return;
	end
	uu.var_dump( ret_table_ );

	--适配ssdb接口 { hash, value, hash, value, hash, value, hash, value, }
	local fret_ = {}
	for i=1, #ret_table_ do
		if  type(ret_table_[i]) == 'table' and ret_table_[i].hash and ret_table_[i].v1 then
			fret_[ #fret_ + 1 ] = ret_table_[i].hash
			fret_[ #fret_ + 1 ] = ret_table_[i].v1
		end
	end

	--log.day_list_simple( "dc_mysql", 'hget_all', uu.to_str(fret_) )

	return fret_
end



--- API 从key>start_开始，遍历size_个key值，最大4096个
_M.API_mysql_hlist = function(self, p_select_, start_, size_, err_call_back)
	log.debug( log.fast() and "call API_mysql_hlist, start=" .. start_ );
	local select_ = p_select_ % self.select_mod

	local start_str_   = uu.check_sql_field_quote( start_ )

	ngx.ctx.err_call_back = err_call_back;
	if  size_ > 4096 then size_ = 4096 end

	local mysql_ = "select k1 from " .. self:private_get_kv_table_name(select_)
			.. " where k1>" .. start_str_ .. " order by k1 limit " .. size_;
	local ret_, ret_table_, err = self:query(mysql_, select_);
	if  not ret_ then
		self:call_mysql_fail( "hlist_fail", start_, size_, err )
		return;
	end
	uu.var_dump(ret_table_);

	--log.day_list_simple( "dc_mysql", 'hlist1', uu.to_str(ret_table_) )

	-- 每一行一个key { 111, 222, 333 } (适配ssdb接口)
	local fret_ = {}
	for i=1, #ret_table_ do
		if  type(ret_table_[i]) == 'table' and ret_table_[i].k1 then
			fret_[#fret_ + 1] = ret_table_[i].k1
		end
	end

	--log.day_list_simple( "dc_mysql", 'hlist', uu.to_str(fret_) )

	return fret_
end



--- API 从id>start_开始，遍历size_个key+hash+value值，最大4096个 （mysql-dc2模式专用，更加先进，使用id字段扫描)
_M.API_mysql_hlist_id = function(self, p_select_, start_, size_, err_call_back)
	log.debug( log.fast() and "call API_mysql_hlist_id, start=" .. start_ );
	local select_ = p_select_ % self.select_mod

	ngx.ctx.err_call_back = err_call_back;
	if  size_ > 4096 then size_ = 4096 end
	local mysql_ = "select id, k1, hash, UNCOMPRESS(v1) as v1 from " .. self:private_get_kv_table_name(select_)
			.. " where id>" .. start_ .. " order by id limit " .. size_;
	local ret_, ret_table_, err = self:query(mysql_, select_);
	if  not ret_ then
		self:call_mysql_fail( "hlist_fail", start_, size_, err )
		return;
	end

	uu.var_dump(ret_table_);
	return ret_table_ or {};
end



--- API 设置一个绑定关系
_M.API_mysql_set_bind = function( self, p_select_, str_, uin_, err_call_back )
	log.debug( log.fast() and "call API_mysql_set_bind" );

	local select_ = p_select_ % self.select_mod
	ngx.ctx.err_call_back = err_call_back;


	--local str_base64_ = ngx.encode_base64( str_ )
	local value_str_ = uu.check_sql_field_quote( str_ )
	local mysql_ = table_concat( { "insert into t_bind_acc_name_uin",
							 "(str,uin)values(", value_str_, ",", uin_, ")",
							 ' on duplicate key update str=', value_str_ } )

	local ret_, ret_table_, err = self:query( mysql_, select_ )
	if  not ret_ then
		if  type(err) == 'string' and  string.sub( err, 1, 15 ) == 'Duplicate entry' then
			return 1, 'Duplicate'
		else
			-- 请求失败
			self:call_mysql_fail( "insert_fail", str_, uin_, err )
			return 1, err
		end
	else
		--{insert_id=0, affected_rows=1, server_status=2, warning_count=0,}
		-- Duplicate entry '中文绑定测试002aaccccca-1001' for key 'i1'
		return 0, "ok"
	end
end



--- API 查询一个绑定关系
_M.API_mysql_get_bind = function( self, p_select_, str_, uin_, err_call_back )
	log.debug( log.fast() and "call API_mysql_get_bind" );

	local select_ = p_select_ % self.select_mod
	ngx.ctx.err_call_back = err_call_back;

	local where_
	if  str_ then
		where_ = 'str=' .. uu.check_sql_field_quote( str_ ) .. ' order by uin'
	elseif tonumber(uin_) then
		where_ = 'uin=' .. uin_ .. ' order by str'
	else
		return 99, "ERROR_PARAMS"
	end

	local mysql_ = table_concat( { "select * from t_bind_acc_name_uin where ", where_, ' limit 1' } )

	local ret_, ret_table_, err = self:query( mysql_, select_ )
	if  not ret_ then
		-- 请求失败，删除掉cache
		self:call_mysql_fail( "insert_fail", str_, uin_, err )
		return 1, err
	else
		--{insert_id=0, affected_rows=1, server_status=2, warning_count=0,}
		-- Duplicate entry '中文绑定测试002aaccccca-1001' for key 'i1'
		return 0, ret_table_
	end
end



--- API 清理掉一个cache数据
_M.API_mysql_clean_shared_cache = function(self, p_select_, key_, hash_ )
	--log.debug( log.fast() and "call API_mysql_clean_shared_cache, select_=" .. p_select_
			--.. ", key=" .. key_ .. ", hash=" .. hash_ );
	local select_ = p_select_ % self.select_mod
	self:clean_shared_cache_mysql(select_, key_, hash_ )
end


--- API 尝试获得远程锁
_M.API_mysql_try_get_resty_lock = function(self, p_select_, key_hash_, et_ )
	local select_ = p_select_ % self.select_mod
	return self:check_mysql_resty_lock(select_, key_hash_, check_lock_expire_time(et_) )
end


--- API 尝试释放远程锁
_M.API_mysql_unlock_resty_lock = function(self, p_select_, key_hash_ )
	local select_ = p_select_ % self.select_mod
	return self:unlock_resty_lock( select_, key_hash_, nil )
end


-- 检查lock_kh的锁id
_M.check_lock_kh_id = function( self, select_, key_hash_, lock_kh_id_ )

	local dict_ = ngx_shared[self.pre .. select_]
	if  dict_ and key_hash_ then
		---检查lock_kh_id
		local lock_id_ = dict_:get( key_hash_ .. "_lk" )
		if  lock_id_ then
			if  lock_id_ == tonumber(lock_kh_id_) then
				-- 验证锁ok
				-- log.day_list_simple( "dc_mysql", "check_lock_kh_id_ok", key_hash_, lock_id_ )
				return 0
			else
				log.day_list_simple( "dc_mysql", "error_check_lock_kh_id",
						key_hash_, lock_id_, lock_kh_id_)
				return 1, "lock_id_not_match"
			end
		else
			log.day_list_simple( "dc_mysql", "error_check_lock_kh_id_not_lock",
					key_hash_, lock_kh_id_ )
			return 2, "not_lock"
		end
	end

	return 99, "sys_err"
end



-----------------------------ngx_shared--------------------------------

--设置全局数据缓存kv 十分表
_M.set_shared_cache_mysql = function( self, select_, key_hash_, str_, expire_, ver_ )
	if  select_ and key_hash_ and str_  then
		local dict_name_ = self.pre ..  select_
		local share_ = ngx_shared[ dict_name_ ];
		if  share_ then
			if  expire_ and type(expire_) == 'string' then
				--log.debug( log.fast() and "expire_=" .. expire_ .. " k=" .. key_ .. " v=" .. str_ )
				expire_ = tonumber(expire_) or 0
			end

			if  expire_ and expire_ >=0 and expire_ < 7200 then
				--normal
			else
				expire_ = 7200     --2*60*60
			end

			local ok, err, forcible = share_:set( key_hash_, str_, expire_ );
			if  ok then
				if  forcible then
					log.day_list_short("dc_mysql", "cache_hit_mysql", "set_forcible", dict_name_, key_hash_, err )
				end

				if  ver_ and ver_ > 0 then
					share_:set( key_hash_ .. __VER__ , ver_, expire_ )   -- 12345__VeR_ 设置数据的版本号
				end

				return true
			else
				if  err then
					--share_:delete( key_ )
					--set出错，立即删除缓存
					local ok, err, forcible = share_:set( key_hash_, nil );
					if  err or forcible then
						log.day_list_short("dc_mysql", "cache_hit_mysql", "error_delete",
								dict_name_, key_hash_, err, forcible )
					end
				end
				log.day_list_short("dc_mysql", "cache_hit_mysql", "error_set", dict_name_, key_hash_, err )
			end
		else
			log.day_list_short("dc_mysql", "cache_hit_mysql", "error_set_no_dict", dict_name_, key_hash_)
		end
	end

	return false   --写入cache失败
end



--获得全局数据缓存kv 十分表
_M.get_shared_cache_mysql = function( self, select_, key_hash_ )
	if  select_ and key_hash_ then
		local dict_name_ = self.pre ..  select_
		local share_ = ngx_shared[ dict_name_ ];
		if  share_ then

			local str_, f_, stale_ = share_:get_stale( key_hash_);
			--uu.report_share_free_space( share_, dict_name_, key_, "dc_mysql" )

			if  str_ then
				return str_, stale_  	--正常返回
			else
				--有报错的情况 ret = nil, error
				if  f_ then
					log.day_list_short("dc_mysql", "cache_hit_mysql", "error_get_nil", dict_name_, key_hash_, f_ )
				end
			end
		else
			log.day_list_short("dc_mysql", "cache_hit_mysql", "error_get_no_dict", dict_name_, key_hash_)
		end
	end

	return nil, nil
end


--清理掉一个cache数据 十分表
_M.clean_shared_cache_mysql = function(self, select_, key_, hash_ )
	if  select_ and key_ and hash_ then
		local dict_name_ = self.pre .. select_
		local share_ = ngx_shared[ dict_name_ ];
		if  share_ then
			local ok, err, forcible = share_:set( key_ .. "_h" .. hash_ , nil );
			if  err or forcible then
				log.day_list_short("dc_mysql", "cache_hit_mysql", "error_delete", dict_name_, key_, err, forcible )
			end
			share_:set( key_ .. "_h" .. hash_ .. __VER__ , nil );
		end
	end
end



--获得全局数据缓存kv表里的ver值 十分表
_M.get_ver_shared_cache_mysql = function( self, select_, key_hash_ )
	if  select_ and key_hash_ then
		local dict_name_ = self.pre ..  select_
		local share_ = ngx_shared[ dict_name_ ];
		if  share_ then
			local str_, f_, stale_ = share_:get_stale( key_hash_ .. __VER__ );
			if  str_ then
				return str_, stale_  	--正常返回
			else
				--有报错的情况 ret = nil, error
				if  f_ then
					log.day_list_short("dc_mysql", "cache_hit_mysql", "error_get_ver_nil",
							dict_name_, key_hash_ .. __VER__, f_ )
				end
			end
		else
			log.day_list_short("dc_mysql", "cache_hit_mysql", "error_get_no_dict", dict_name_, key_hash_ .. __VER__ )
		end
	end

	return nil, nil
end



return _M;
